/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.operators;

import java.io.BufferedReader;
import java.io.FileReader;
import java.nio.ByteBuffer;
import java.util.regex.Pattern;

import edu.ucr.cs.dblab.acestore.api.IOperator;
import edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory;
import edu.ucr.cs.dblab.acestore.api.TupleInterpreter;
import edu.ucr.cs.dblab.acestore.base.ACEContext;
import edu.ucr.cs.dblab.acestore.base.datatypes.ISerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.IntegerSerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;

/**
 * 
 */
public class FileScanOperatorFactory implements ISinglePrevOperatorFactory {

    private final SchemaDescriptor inputSchemaDescriptor;

    private final String delimiter;

    private final String filepath;

    public FileScanOperatorFactory(String filepath,
            SchemaDescriptor inSchemaDescriptor, String delimiter) {
        this.inputSchemaDescriptor = inSchemaDescriptor;
        this.delimiter = delimiter;
        this.filepath = filepath;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * edu.ucr.cs.dblab.acestore.api.IOperatorFactory#createOperator(edu.ucr
     * .cs.dblab.acestore.base.ACEContext,
     * edu.ucr.cs.dblab.acestore.api.IOperator)
     */
    @Override
    public IOperator createOperator(ACEContext ctx, IOperator prevOperator) {

        final Pattern pattern = Pattern.compile(delimiter);
        final TupleInterpreter tupleBuilder = new TupleInterpreter(
                inputSchemaDescriptor.getFields().length);

        return new IOperator() {

            BufferedReader breader;

            private byte[] tupleBuf;

            private boolean isTupleLeft;

            private int outPageID;

            @Override
            public void open() throws Exception {
                breader = new BufferedReader(new FileReader(filepath));
                isTupleLeft = false;
                this.outPageID = 0;
            }

            @Override
            public void getNext(ByteBuffer page) throws Exception {
                int count = 0;
                if (isTupleLeft) {
                    if (page.position() + tupleBuf.length
                    // Additional three fields: pageRef, recCount, recEndOffset
                            + 3 * INTSIZE - RESERVED_END >= page.capacity()) {
                        throw new Exception("Cannot put a record into a page!");
                    }
                    page.put(tupleBuf);
                    isTupleLeft = false;
                    count++;
                }
                String line;
                while ((line = breader.readLine()) != null) {
                    if(line.trim().isEmpty())
                        continue;
                    tupleBuilder.reset();
                    String[] fields = pattern.split(line);
                    for (int i = 0; i < inputSchemaDescriptor.getFields().length; i++) {
                        tupleBuilder
                                .addField(inputSchemaDescriptor.getFields()[i]
                                        .serialize(fields[i]));
                    }
                    tupleBuf = tupleBuilder.buildTuple();
                    if (page.position() + tupleBuf.length + INTSIZE >= page
                            .capacity()
                            - 2
                            * INTSIZE
                            - count
                            * INTSIZE
                            - RESERVED_END) {
                        isTupleLeft = true;
                        page.putInt(page.capacity() - INTSIZE - RESERVED_END,
                                outPageID + 1);
                        outPageID++;
                        return;
                    } else {
                        // put record into the buffer
                        page.put(tupleBuf);
                        count++;
                        // End offset of the tuple
                        page.putInt(page.capacity() - 2 * INTSIZE - count
                                * INTSIZE - RESERVED_END, page.position() - 1);
                        // Count of tuples in the page
                        page.putInt(page.capacity() - 2 * INTSIZE
                                - RESERVED_END, count);
                    }
                }
                // Set the last page reference to be -1
                page.putInt(page.capacity() - 2 * INTSIZE - RESERVED_END, count);
                page.putInt(page.capacity() - INTSIZE - RESERVED_END, -1);
            }

            @Override
            public void close() throws Exception {
                breader.close();
                tupleBuf = null;
            }
        };
    }

    public static void main(String[] args) throws Exception {
        FileScanOperatorFactory scanOperatorFactory = new FileScanOperatorFactory(
                "/Users/jarodwen/Desktop/hyracks/data/simple2D.txt",
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), ",");
        IOperator scanOperator = scanOperatorFactory.createOperator(null, null);
        byte[] page = new byte[100];
        ByteBuffer buf = ByteBuffer.wrap(page);
        scanOperator.open();
        scanOperator.getNext(buf);
        while (buf.getInt(buf.capacity() - 4) >= 0) {
            System.out.println(buf.toString());
            buf.position(0);
            scanOperator.getNext(buf);
        }
    }
}
